In [1]:
from pyspark.sql import SparkSession #this is the module that connects to apache spark
from pyspark.sql import SQLContext
In [2]:
spark = (SparkSession.builder.appName("csp1_jes").config("hive.metastore.uris", "thrift://ip-10-1-2-24.ap-south-1.compute.internal:9083").enableHiveSupport().getOrCreate())
#this creates a Spark session that allows us to connect to spark and configure the program
In [3]:
spark
Out[3]:
Employee Data¶
- the database and table structureIn [4]:
#The database for my porject is named 'csp1_jes'
spark.sql("use csp1_jes")
Out[4]:
In [5]:
#List of all tables in my database
spark.sql("show tables").show()
In [6]:
#The desc shows the description of the schema of the table
#Titles - Job titles of employees
spark.sql("desc titles_jes").show()
In [7]:
spark.sql("Select * from titles_jes").show()
In [8]:
#Departments - The departments in the organisation
spark.sql("desc departments_jes").show()
In [9]:
spark.sql("Select * from departments_jes").show()
In [10]:
#Department Employees - which department each employee belongs to
spark.sql("desc dept_emp_jes").show()
In [11]:
spark.sql("Select * from dept_emp_jes limit 20").show()
In [12]:
#Department Managers - Managers of each department in the organisation
spark.sql("desc dept_manager_jes").show()
In [13]:
spark.sql("select * from dept_manager_jes limit 20").show()
In [14]:
#Employees - Personal employee information
spark.sql("desc employees_jes").show()
In [15]:
spark.sql("select * from employees_jes limit 20").show()
In [16]:
#Salaries - salaries of each employee
spark.sql("desc salaries_jes").show()
In [17]:
spark.sql("select * from salaries_jes limit 20").show()
1) A list showing employee number, last name, first name, sex, and salary for each employee1.¶
In [18]:
spark.sql("select e.emp_no, e.last_name, e.first_name, e.sex, s.salary from employees_jes e JOIN salaries_jes s on e.emp_no = s.emp_no").show()
2) A list showing first name, last name, and hire date for employees who were hired in 1986.¶
In [19]:
spark.sql("select first_name, last_name, hire_date from employees_jes where hire_date between '1/1/1986' and '12/31/1986' order by hire_date").show()
3) A list showing the manager of each department with the following information: department number, department name, the manager's employee number, last name, first name.¶
In [20]:
spark.sql("select d.dept_no, d.dept_name, dm.emp_no, e.last_name, e.first_name from departments_jes d join dept_manager_jes dm on d.dept_no = dm.dept_no join employees_jes e on dm.emp_no = e.emp_no").show()
4) A list showing the department of each employee with the following information: employee number, last name, first name, and department name.¶
In [21]:
spark.sql("select de.emp_no, e.last_name, e.first_name, d.dept_name from dept_emp_jes de join employees_jes e on de.emp_no = e.emp_no join departments_jes d on de.dept_no = d.dept_no").show()
5) A list showing first name, last name, and sex for employees whose first name is "Hercules" and last names begin with "B".¶
In [22]:
spark.sql("select first_name, last_name, sex from employees_jes where first_name = 'Hercules' and last_name Like 'B%'").show()
6) A list showing all employees in the Sales department, including their employee number, last name, first name, and department name.¶
In [23]:
spark.sql("select e.emp_no, e.last_name, e.first_name, d.dept_name from employees_jes e join dept_emp_jes de \
on de.emp_no = e.emp_no join departments_jes d on de.dept_no = d.dept_no where d.dept_no like '%d007%'").show()
7) A list showing all employees in the Sales and Development departments, including their employee number, last name, first name, and department name.¶
In [24]:
spark.sql("select e.emp_no, e.last_name, e.first_name, d.dept_name from employees_jes e join dept_emp_jes de \
on de.emp_no = e.emp_no join departments_jes d on de.dept_no = d.dept_no where d.dept_no like '%d007%' or d.dept_no like '%d005%'").show()
8) A list showing the frequency count of employee last names, in descending order. (i.e., how many employees share the same last name)¶
In [25]:
spark.sql("select last_name, COUNT(last_name) as lm_frequency from employees_jes group by last_name order by lm_frequency DESC").show()
9) Histogram to show the salary distribution among the employees¶
In [26]:
import numpy as np #Basically a fast linear algebra library
import seaborn as sns #Data visualisation based on matplotlib
import matplotlib.pyplot as plt #Data visualisation
import pandas as pd #To convert hive data into a pandas dataframe for data visualization
In [27]:
#Creating a pandas dataframe with salary column
emp_salaries = spark.sql("select * from salaries_jes")
In [28]:
emp_salaries = emp_salaries.toPandas()
In [29]:
emp_salaries.head(20)
Out[29]:
In [30]:
#Statistical overview of the salaries
emp_salaries['salary'].describe()
Out[30]:
In [30]:
#pip install -U seaborn
In [31]:
plt.figure(figsize = (15, 10))
sns.histplot(data = emp_salaries, x = 'salary', bins = 10, kde = True, color = '#FB4E0B')
plt.title('Employee Salary Distribution', fontsize = 18)
plt.xlabel('Employee Salary', fontsize = 16)
plt.ylabel('Employee count', fontsize = 16)
Out[31]:
10) Bar graph to show the Average salary per title (designation)¶
In [32]:
#Creating pandas dataframe for titles
emp_titles = spark.sql("select * from titles_jes")
In [33]:
emp_titles = emp_titles.toPandas()
In [34]:
emp_titles.head()
Out[34]:
In [35]:
#Creating pandas dataframe for employees
emp = spark.sql("select * from employees_jes")
In [36]:
emp = emp.toPandas()
In [37]:
emp.head()
Out[37]:
In [38]:
dept_emp = spark.sql("select * from dept_emp_jes")
In [39]:
dept_emp = dept_emp.toPandas()
In [40]:
dept_emp.head()
Out[40]:
In [41]:
#Rename columns to merge tables
emp_new = emp_titles.rename(columns={'title_id': 'emp_title_id'})
In [42]:
emp_new.head()
Out[42]:
In [43]:
#Merge salaries and employees tables
emp_comb = pd.merge(emp_salaries, emp, on="emp_no", how="inner")
In [44]:
emp_comb.head()
Out[44]:
In [45]:
emp_comb2 = pd.merge(emp_comb, emp_new, on="emp_title_id", how="inner")
In [46]:
emp_comb2.head()
Out[46]:
In [47]:
emp_sal_grouped = emp_comb2.groupby("title").mean()
In [48]:
emp_sal_grouped
Out[48]:
In [49]:
#In emp_sal_grouped the index is the title therefor it needs to be changed to use for plotting later on
emp_sal_grouped2 = emp_sal_grouped.reset_index()
In [50]:
emp_sal_grouped2
Out[50]:
In [51]:
emp_sal_grouped2 = emp_sal_grouped2.sort_values('salary')
In [52]:
plt.figure(figsize = (15, 10))
sns.barplot(x = 'salary', y = 'title', data = emp_sal_grouped2, palette = 'rocket_r')
plt.title('Employee Salary Distribution', fontsize = 18)
plt.xlabel('Employee Salary', fontsize = 16)
plt.ylabel('Employee Job Title', fontsize = 16)
Out[52]:
Insights¶
- It is apparent that senior staff earns the highest salary followed by Managers and lastly other staffIn [53]:
#Making one dataframe from all separate data files
emp_full = pd.merge(dept_emp, emp_comb2, on="emp_no", how="inner")
In [54]:
emp_full.head()
Out[54]:
11) Salary distribution by sex¶
In [55]:
import plotly.express as px
In [56]:
plt.figure(figsize = (15, 10))
plot1 = px.histogram(emp_full, x = 'salary', color= "sex", nbins = 10, title='Salary distribution by sex',
labels={'salary':'Employee Salary'})
plot1.show()
12) No of employees per designation segregated by salary¶
In [57]:
plt.figure(figsize = (15, 10))
plot2 = px.bar(emp_sal_grouped2, x = 'title', y = 'emp_no', color = 'salary')
plot2.show()
Machine Learning¶
In [58]:
#Making a copy so that I can use it if there are any errors in future code
emp_full2 = emp_full.copy()
In [59]:
#Dataframe information
emp_full2.info()
In [60]:
#Change the column types accordingly for further use
emp_full2['hire_date'] = pd.to_datetime(emp_full2['hire_date'])
In [61]:
emp_full2['birth_date'] = pd.to_datetime(emp_full2['birth_date'])
In [62]:
#The left_org column is the column that gives information about when an employee left the job post.
#It has not been stored in a proper format hence I need to transform it.
emp_full2['last_date']
Out[62]:
In [63]:
emp_full2['last_date'] = emp_full2['last_date'].replace(to_replace = '\r', value = '', regex = True)
In [64]:
emp_full2['last_date']
Out[64]:
In [65]:
emp_full2['last_date'] = pd.to_datetime(emp_full2['last_date'])
In [66]:
emp_full2.info()
There are total of 7 categorical variables.
- dept_no
- emp_title_id
- first_name
- last_name
- sex
- last_performance_rating
- title
But not all are important for ML model, hence I will only transform those that are important which are emp_title_no/title, dept_no, sex, last_performace_rating. Here the dimension of columns is less and easily manageable hence I will encode thhe columns manually instead of using encoding functions and increasing the dimensions
In [67]:
emp_full2['sex'] = np.where(emp_full2['sex'] == 'F', 0, 1)
In [68]:
emp_full2['last_performance_rating'].value_counts()
Out[68]:
The last performance rating is a ordinal variable where the ratings are in the order of the highest rating being S, A, B, C and the least rating being PIP. While encoding ordinal variable the usual convention is to give highest value to the highest ordinal value. Hence I will give the ratings are follows:
- S -> 4
- A -> 3
- B -> 2
- C -> 1
- PIP -> 0In [69]:
emp_full2['last_performance_rating'] = emp_full2['last_performance_rating'].replace(to_replace = 'S', value = 4)
In [70]:
emp_full2['last_performance_rating'] = emp_full2['last_performance_rating'].replace(to_replace = 'A', value = 3)
In [71]:
emp_full2['last_performance_rating'] = emp_full2['last_performance_rating'].replace(to_replace = 'B', value = 2)
In [72]:
emp_full2['last_performance_rating'] = emp_full2['last_performance_rating'].replace(to_replace = 'C', value = 1)
In [73]:
emp_full2['last_performance_rating'] = emp_full2['last_performance_rating'].replace(to_replace = 'PIP', value = 0)
In [74]:
emp_full2['last_performance_rating'].value_counts()
Out[74]:
In [75]:
emp_full2['emp_title_id'].value_counts()
Out[75]:
Employee title id doesnt have any order hence, it doesn't matter what numbers are used to encode.
In [76]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 's0001', value = 0)
In [77]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 's0002', value = 1)
In [78]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 'e0001', value = 2)
In [79]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 'e0002', value = 3)
In [80]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 'e0003', value = 4)
In [81]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 'e0004', value = 5)
In [82]:
emp_full2['emp_title_id'] = emp_full2['emp_title_id'].replace(to_replace = 'm0001', value = 6)
In [83]:
emp_full2['emp_title_id'].value_counts()
Out[83]:
In [84]:
emp_full2['dept_no'].value_counts()
Out[84]:
There are a total of nine departments and they are not ordinal. Hence it does not matter what the numbers are used to encode.
In [85]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd001', value = 0)
In [86]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd002', value = 1)
In [87]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd003', value = 2)
In [88]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd004', value = 3)
In [89]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd005', value = 4)
In [90]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd006', value = 5)
In [91]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd007', value = 6)
In [92]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd008', value = 7)
In [93]:
emp_full2['dept_no'] = emp_full2['dept_no'].replace(to_replace = 'd009', value = 8)
In [94]:
emp_full2['dept_no'].value_counts()
Out[94]:
In [95]:
emp_full2.head()
Out[95]:
In [96]:
plt.figure(figsize = (15, 10))
null_plot = sns.heatmap(emp_full2.isnull(), cbar = False)
plt.title('Null value distribution', fontsize = 18)
plt.xlabel('Variables', fontsize = 16)
plt.ylabel('Count', fontsize = 16)
Out[96]:
In [97]:
emp_full2.isna().sum()
Out[97]:
Insights¶
- There are no null values in any variables except 'last_date' which means only a few employees have left the organisationIn [98]:
emp_full2.columns
Out[98]:
In [99]:
#Correlation plot to find out between variables in the data
corr_features = ['dept_no', 'salary', 'emp_title_id', 'birth_date', 'sex', 'hire_date', 'no_of_projects',
'last_performance_rating', 'left_org', 'last_date']
In [100]:
corr1 = emp_full2[corr_features]
In [101]:
corr1.head()
Out[101]:
In [102]:
plt.figure(figsize = (25, 20))
corr_plot = sns.heatmap(corr1.corr(), annot = True, fmt = ".2%", cmap = 'Oranges', annot_kws = {'size': 8})
plt.title('Correlation betweena all variables')
plt.show()
Insights¶
- There doesn't seem to be much correlation between the variables therefore I will keep all of them for the model buildingIn [103]:
emp_spark = spark.createDataFrame(emp_full2)
In [104]:
emp_spark.printSchema()
In [105]:
emp_spark.show()
In [106]:
#Creating a dataframe with the relevant variables
emp_mfeatures = emp_spark.select('dept_no', 'salary', 'emp_title_id', 'birth_date', 'sex', 'hire_date', 'no_of_projects',
'last_performance_rating', 'last_date', 'left_org')
In [107]:
emp_mfeatures.printSchema()
In [108]:
#converting the datatypes to spark datatypes
from pyspark.sql.types import StringType, DateType, IntegerType
In [109]:
emp_mfeatures = emp_mfeatures.withColumn("birth_date",
emp_mfeatures["birth_date"]
.cast(IntegerType()))
In [110]:
emp_mfeatures = emp_mfeatures.withColumn("hire_date",
emp_mfeatures["hire_date"]
.cast(IntegerType()))
In [111]:
emp_mfeatures = emp_mfeatures.withColumn("last_date",
emp_mfeatures["last_date"]
.cast(IntegerType()))
In [112]:
emp_mfeatures = emp_mfeatures \
.withColumn("dept_no" ,
emp_mfeatures["dept_no"]
.cast(IntegerType())) \
.withColumn("salary",
emp_mfeatures["salary"]
.cast(IntegerType())) \
.withColumn("emp_title_id" ,
emp_mfeatures["emp_title_id"]
.cast(IntegerType())) \
.withColumn("sex" ,
emp_mfeatures["sex"]
.cast(IntegerType())) \
.withColumn("no_of_projects",
emp_mfeatures["no_of_projects"]
.cast(IntegerType())) \
.withColumn("last_performance_rating",
emp_mfeatures["last_performance_rating"]
.cast(IntegerType())) \
.withColumn("left_org",
emp_mfeatures["left_org"]
.cast(IntegerType())) \
In [113]:
emp_mfeatures.printSchema()
In [114]:
features = emp_mfeatures.columns
In [115]:
emp_list = emp_mfeatures.columns[0:11]
In [116]:
#using the assembler to create a single vector column from a list of columns
from pyspark.ml.feature import VectorAssembler
In [117]:
assemblerInputs = emp_list
vecAssembler = VectorAssembler(inputCols = assemblerInputs, outputCol = "features")
In [118]:
#Definig the model
from pyspark.ml.classification import LogisticRegression
In [119]:
emp_lr = LogisticRegression(featuresCol = "features", labelCol = "left_org", regParam = 1.0)
In [120]:
#Building a pipeline to create the model
from pyspark.ml import Pipeline
In [121]:
#Defining the pipeline based on the stages created in previous steps
emp_pipeline = Pipeline(stages = [vecAssembler, emp_lr])